package com.shengzai.rdd

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo17Action {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("join")

    val sc = new SparkContext(conf)

    println(1)
    val studentRDD: RDD[String] = sc.textFile("data/students.txt")
    println(2)
    /**
     * 转换算子是懒执行，必须有一个action算在内触发任务执行，
     * 整个处理数据的代码想象管道，数据再管道中流通，一条一条按照数据处理的
     * 会再shuffle的位置落地到磁盘
     */
    val clazzsRDD: RDD[String] = studentRDD.map(line => {
      println("map执行")
      val split: Array[String] = line.split(",")
      split.last
    })
    println(3)

    /**
     * action算子L:触发任务执行，每一个action算子都会触发一个job执行，会将前面的代码执行一边
     *
     */

    /**
     * foreach action算子，触发任务的执行
     */
    clazzsRDD.foreach(println)
    println(4)

    /**
     * count: 统计rdd总数
     *
     */

    val count: Long = clazzsRDD.count()

    println(count)

    /**
     * collect ：将RDD转换成scala集合
     *
     */

    val array: Array[String] = clazzsRDD.collect()

    array.foreach(println)

    /**
     * reduce: 全局聚合
     *
     */

    val classStr: String = clazzsRDD.reduce((x, y) => x + y)

    println(classStr)


    /**
     * take: 取top，返回一个数组
     */

    val takeArray: Array[String] = clazzsRDD.take(100)

    takeArray.foreach(println)

    /**
     * save:保存结果
     *
     */

    clazzsRDD.saveAsTextFile("data/clazzs")
  }

}
